package com.atguigu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;

import com.atguigu.bean.TableProcess;
import com.atguigu.common.GmallConfig;
import com.atguigu.util.DruidDSUtil;
import com.atguigu.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;


public class DimTableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject>{
    // 添加状态描述器
    public MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    // 定义 Phoenix 连接对象  德鲁伊连接
    private DruidDataSource druidDataSource;
    // 解决主流先来丢数据的问题，用预加载
    private HashMap<String, TableProcess> tableProcessHashMap;

    public DimTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化连接池,然后可以去建表
        druidDataSource = DruidDSUtil.getDruidDataSource();

        // new 一个 HashMap<>
        tableProcessHashMap = new HashMap<>();
        Connection connection = DriverManager.getConnection( "jdbc:mysql://47.102.112.46:3306/edu_config?user=root&password=qw123456&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false" );
        List<TableProcess> tableProcesses = JdbcUtil.queryList( connection, "select * from table_process where sink_type='dim' ", TableProcess.class, true );

        // 将集合数据封装进 Map 建表
        for (TableProcess tableProcess : tableProcesses) {
            checkTable( tableProcess.getSinkTable(),
                    tableProcess.getSinkPk(),
                    tableProcess.getSinkColumns(),
                    tableProcess.getSinkExtend() );
            tableProcessHashMap.put( tableProcess.getSourceTable(), tableProcess );
        }
        connection.close();
    }

    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject( value );

        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState( mapStateDescriptor );
        // 0. 如果当前为删除数据操作，则从状态中也要将数据删除 具体表现为
        // 如果操作状态为删除d 就把它移除掉

        if ("d".equals( jsonObject.getString( "op" ) )) {
            // 如果是 d 删除操作，那 after里是空的，所以这里应该取 before的表名。
            String key = jsonObject.getJSONObject( "before" ).getString( "source_table" );

            broadcastState.remove( key );
            tableProcessHashMap.remove( key );// hashMap 可以删掉一个不存在的元素，不会报错
        } else {
            String key = jsonObject.getJSONObject( "after" ).getString( "source_table" );

            // todo 注意：主流数据删了要删除状态中的数据吗？没必要，因为删了mysql的数据，后面是访问不到的，没必要删
            // 1. 解析数据为 TableProcess 对象
            TableProcess tableProcess = JSONObject.parseObject( jsonObject.getString( "after" ), TableProcess.class );

            // 2. 校验并建表    sql建表语句
            if ("dim".equals( tableProcess.getSinkType() )){
                checkTable( tableProcess.getSinkTable(),
                        tableProcess.getSinkPk(),
                        tableProcess.getSinkColumns(),
                        tableProcess.getSinkExtend() );

                // 3. 将数据写入状态
//            broadcastState.put( tableProcess.getSourceTable(),tableProcess );
                broadcastState.put( key, tableProcess ); // 两个效果一样
            }

        }
    }

    // 校验并建表：create table if not exists db.tn(id varchar primary key,name varchar,sex varchar) xxx;
    private void checkTable(String sinkTable, String sinkPK, String sinkColumns, String sinkExtend) {
        DruidPooledConnection connection = null;
        PreparedStatement preparedStatement = null;

        try {  // 如果没有主键，需要给 sinkPK 一个默认值
            if (sinkPK == null || "".equals( sinkPK )) {
                sinkPK = "id";
            } // 扩展字段也有可能为null 也要处理 防止null报错，可以给一个空字符串
            if (sinkExtend == null) {
                sinkExtend = "";
            }
            // 拼接建表sql语句
            StringBuilder sql = new StringBuilder( "create table if not exists " )
                    .append( GmallConfig.PHOENIX_DB )
                    .append( "." )
                    .append( sinkTable )
                    .append( "(" );
            // 切分字段，“id，name，sex”
            String[] fields = sinkColumns.split( "," );
            for (int i = 0; i < fields.length; i++) {
                String field = fields[i];
                sql.append( field ).append( " varchar" );
                // 判断是否为主键字段 如果是主键需要添加  primary key
                if (sinkPK.equals( field )) {
                    sql.append( " primary key" );
                }
                // 判断是否为最后一个字段，如果不是最后一个字段，则需要添加 逗号 ,
                if (i < fields.length - 1) {
                    sql.append( "," );
                }
            }
            sql.append( ")" )
                    .append( sinkExtend );

            // 打印 sql 语句
            System.out.println( "打印sql语句，检查sql语句有无错误：" + sql );
            // 获取连接
            connection = druidDataSource.getConnection();
            // 执行 sql 语句
            preparedStatement = connection.prepareStatement( sql.toString() );

            //执行SQL
            preparedStatement.execute();

        } catch (SQLException e) {
            e.printStackTrace();
            // todo 化编译时异常为运行时异常，把它抛出去在外面解决掉
            throw new RuntimeException( "建表：" + sinkTable +
                    "失败 ！！！建表失败任务要停止！！！" +
                    "数据来了也写不进去，所以这里的异常可以 try catch ==也可以抛出去  " +
                    "记得关闭流，否则会占用掉所有的资源" );
        }
        //这里是在建表，如果建表失败，哪怕数据来了也写不进去
        // 所以这里的异常可以try catch {也可以抛出去}
        // 流必须关闭，否则会占用掉所有的资源
        try {
            if (preparedStatement != null) {
                preparedStatement.close();
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }

        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }


    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //主流 读的是topic里所有的数据 如果事实表来了，很有可能为null
        // 1.读状态 按照 key取值，也就是要从value中取到 source_table
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState( mapStateDescriptor );
        String key = value.getString( "table" );
        // 获取配置信息
        TableProcess tableProcess = broadcastState.get( key );
        TableProcess tableProcessMap = tableProcessHashMap.get( key );

        JSONObject data = value.getJSONObject( "data" );  //aa

//        // 2.过滤数据（行，列级别）数据有可能为 null
//        if (tableProcess != null){ // 行过滤
//            // 如果行过滤不为null{key} 再进列过滤
//            filterColumns(value.getJSONObject( "data" ),tableProcess.getSinkColumns());

        // 补充！！！如果是全量表过来，maxwell会有 Bootstrap-start 和 Bootstrap-complete 的脏数据，data为空
        // 2.过滤数据（行，列级别）数据有可能为 null   //bb
        String type = value.getString( "type" );
        if ((tableProcess != null || tableProcessMap != null) && ("insert".equals( type ) || "update".equals( type ) || "bootstrap-insert".equals( type ))) { // 行过滤
            // 补充：不可能同时为null，如果tableProcess为null，可以把tableProcessMap给它。
            if (tableProcess == null) {
                tableProcess = tableProcessMap;
            }
            // 如果行过滤不为null{key} 再进列过滤
            filterColumns( data, tableProcess.getSinkColumns() );

            // todo  3. 将过滤后的数据输出???????????
            // 添加输出的表名，下一步才能用
            value.put( "sink_table", tableProcess.getSinkTable() );
            out.collect( value );

            // 3.如果tableProcess不为null，将过滤后的数据输出
        } else {
            // else 里都是 tableProcess == null
            System.out.println( "没有对应" + key + "的配置信息！" );
        }
    }
    private void filterColumns(JSONObject data, String sinkColumns) {
        // 遇到json就把它当做map处理
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();

        // 迭代配置表里的字段，先切分一下，否则会有包含bug，不符合的就 remove掉
        String[] fields = sinkColumns.split( "," );
        List<String> fieldList = Arrays.asList( fields );

        //迭代，不要的数据remove掉  list是否包含
        while (iterator.hasNext()) {
            Map.Entry<String, Object> entry = iterator.next();
            if (!fieldList.contains( entry.getKey() )) {
                iterator.remove();
            }
        }
    }
}
